package Netty.Reactor模型;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 多路复用，reactor线程模型
 */

public class NioReactor2 {

    abstract class ReactorThread extends Thread {

        //创建一个Selector
        public Selector selector;
        //用于判断线程是否已经启动
        public volatile boolean isRunning = false;

        /**
         * 有事件发生，就调用这个方法
         */
        public abstract void handler(SelectableChannel channel) throws IOException;

        public ReactorThread() throws IOException {
            selector = Selector.open();
        }

        @Override
        public void run() {
            while (isRunning) {
                //启动Selector
                try {
                    selector.select();
                    //获取事件
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    //遍历事件
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        int readyOps = key.readyOps();
                        //只关注是否有OP_ACCEPT和OP_READ事件
                        if ((readyOps & (SelectionKey.OP_ACCEPT | SelectionKey.OP_READ)) != 0 || readyOps == 0) {
                            try {
                                //获取channel
                                SelectableChannel channel = key.channel();
                                //设置为非阻塞模式
                                channel.configureBlocking(false);
                                //有事件，就调用handler方法
                                handler(channel);
                                if (!channel.isOpen()) {
                                    //如果channel关闭，就取消key
                                    key.cancel();
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                //如果有异常，就取消key
                                key.cancel();
                            }
                        }
                        //处理完之后要移除
                        iterator.remove();
                    }
                    selector.selectNow();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        public SelectionKey register(SelectableChannel channel) throws ClosedChannelException {
            //先注册到Selector，并没有注册任何事件
            return channel.register(selector, 0);

        }

        @Override
        public void start() {
            //判断SubReactor线程是否已经启动
            //如果没有启动，就启动SubReactor线程
            if (!isRunning) {
                isRunning = true;
                super.start();
            }
        }
    }

    /**
     * 服务端通道
     */
    public ServerSocketChannel serverSocketChannel;
    /**
     * 用来接收客户端连接
     */
    public ReactorThread[] mainReactors = new ReactorThread[1];

    /**
     * 用来处理客户端连接的读取、写入
     */
    public ReactorThread[] subReactors = new ReactorThread[10];

    /**
     * 线程池，用来处理客户端连接后的业务逻辑
     */
    public ExecutorService threadPool = Executors.newCachedThreadPool();

    /**
     * 初始化mainReactors和subReactors
     */
    public void initAndRegister() throws IOException {
        //subReactors线程，用来客户端连接后的读写
        for (int i = 0; i < subReactors.length; i++) {
            subReactors[i] = new ReactorThread() {
                @Override
                public void handler(SelectableChannel channel) throws IOException {
                    SocketChannel socketChannel = (SocketChannel) channel;
                    ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                    while (socketChannel.isOpen() && socketChannel.read(readBuffer) != -1) {
                        //如果有数据可读，简单的判断一下大于0
                        if (readBuffer.position() > 0) {
                            break;
                        }
                    }
                    //没有数据可读，就直接返回
                    if (readBuffer.position() == 0) {
                        return;
                    }
                    //转换为读取模式
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.limit()];
                    readBuffer.get(bytes);
                    System.out.println("获取到新的数据：" + new String(bytes));
                    System.out.println("获取到新的数据，来自：" + socketChannel.getRemoteAddress());
                    //线程池，用来处理业务数据
                    threadPool.execute(new Runnable() {
                        @Override
                        public void run() {
                        }
                    });
                    //向客户端写数据
                    String response = "HTTP/1.1 200 OK\r\n" + "Content-Length: 11\r\n\r\n" + "hello world";
                    ByteBuffer writeBuffer = ByteBuffer.wrap(response.getBytes());
                    while (writeBuffer.hasRemaining()) {
                        socketChannel.write(writeBuffer);
                    }
                }
            };
        }

        //mainReactors线程，用于客户端的连接
        for (int i = 0; i < mainReactors.length; i++) {
            mainReactors[i] = new ReactorThread() {
                AtomicInteger integer = new AtomicInteger(0);

                @Override
                public void handler(SelectableChannel channel) throws IOException {
                    //获取客户端通道
                    SocketChannel socketChannel = ((ServerSocketChannel) channel).accept();
                    //设置为非阻塞模式
                    socketChannel.configureBlocking(false);
                    int index = integer.getAndIncrement() % subReactors.length;
                    ReactorThread subReactor = subReactors[index];
                    //启动线程
                    subReactor.start();
                    //注册事件
                    SelectionKey key = subReactor.register(socketChannel);
                    key.interestOps(SelectionKey.OP_READ);
                    System.out.println("收到新连接：" + socketChannel.getRemoteAddress());
                }
            };
        }

    }

    /**
     * 初始化服务端
     */

    public void init() throws IOException {
        //创建一个服务端通道
        serverSocketChannel = ServerSocketChannel.open();
        //设置为非阻塞模式
        serverSocketChannel.configureBlocking(false);
        //注册到mainReactor-Thread
        int index = new Random().nextInt(mainReactors.length);
        SelectionKey keys = mainReactors[index].register(serverSocketChannel);
        keys.interestOps(SelectionKey.OP_ACCEPT);
        //启动mainReactor-Thread线程
        mainReactors[index].start();

    }

    /**
     * 服务端绑定端口
     */

    public void bind() throws IOException {
        serverSocketChannel.socket().bind(new InetSocketAddress(8056));
        System.out.println("服务端启动成功");
    }

    public static void main(String[] args) throws IOException {
        NioReactor2 nioReactor = new NioReactor2();
        nioReactor.initAndRegister();
        nioReactor.init();
        nioReactor.bind();
    }

}